package org.apache.flink.runtime.metrics.scope;

import org.apache.flink.runtime.metrics.groups.ResourceProviderConsumerMetricGroup;
import org.apache.flink.runtime.metrics.groups.ResourceSlotMetricGroup;
import org.apache.flink.runtime.resourceconsumer.ConsumeAttemptID;

/** The scope format for the {@link ResourceSlotMetricGroup}. */
public class ResourceSlotScopeFormat extends ScopeFormat {

    public ResourceSlotScopeFormat(String format, ResourceProviderConsumerScopeFormat parentFormat) {
        super(
                format,
                parentFormat,
                new String[] {
                    SCOPE_HOST,
                    SCOPE_TASKMANAGER_ID,
                    SCOPE_JOB_ID,
                    SCOPE_JOB_NAME,
                    SCOPE_TASK_ATTEMPT_ID,
                    SCOPE_TASK_NAME,
                    SCOPE_TASK_ATTEMPT_NUM
                });
    }

    public String[] formatScope(
            ResourceProviderConsumerMetricGroup parent,
            ConsumeAttemptID attemptId,
            String taskName,
            int attemptNumber) {

        final String[] template = copyTemplate();
        final String[] values = {
            parent.parent().hostname(),
            parent.parent().taskManagerId(),
            valueOrNull(parent.jobId()),
            valueOrNull(parent.consumerName()),
            valueOrNull(attemptId),
            valueOrNull(taskName),
            String.valueOf(attemptNumber)
        };
        return bindVariables(template, values);
    }
}
